package cn.doitedu.api;

import beans.UserAction;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/24
 * @Desc: 学大数据，上多易教育
 * TTL: time to live (存活时长)
 **/
public class _14_StateTTL_Demo {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> stream = env.socketTextStream("doitedu", 8899);

        SingleOutputStreamOperator<UserAction> beans = stream.map(json -> JSON.parseObject(json, UserAction.class));


        beans.keyBy(bean -> bean.getUid())
                .process(new KeyedProcessFunction<Integer, UserAction, String>() {


                    ListState<UserAction> listState;

                    @Override
                    public void open(Configuration parameters) throws Exception {


                        ListStateDescriptor<UserAction> desc = new ListStateDescriptor<>("action-list", UserAction.class);
                        // ttl配置  : ttl是对状态容器中的每条数据独立计时
                        StateTtlConfig ttlConfig =
                                new StateTtlConfig.Builder(Time.seconds(2))
                                .updateTtlOnReadAndWrite() // ttl的计时在数据被读或更新之后，重新计算
                                //.updateTtlOnCreateAndWrite() // ttl的计时在数据被放入或更新之后，重新计算
                                .neverReturnExpired()  // 从状态中读取数据时，不返回已过期的数据
                                .build();

                        // 开启ttl
                        desc.enableTimeToLive(ttlConfig);

                        listState = getRuntimeContext().getListState(desc);

                    }

                    @Override
                    public void processElement(UserAction userAction, KeyedProcessFunction<Integer, UserAction, String>.Context ctx, Collector<String> out) throws Exception {

                        listState.add(userAction);

                        int i = 0;
                        for (UserAction action : listState.get()) {
                            i++;
                        }

                        out.collect(ctx.getCurrentKey() + ",状态中数据条数: " + i);

                    }
                }).print();


        env.execute();

    }

}
